package cn.jark.elasticjob;

import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;

import java.lang.annotation.Annotation;
import java.util.Arrays;

/**
 * @author 雨一直夏灬(须知少年凌云志, 曾许人间第一流)
 * @since 2020-10-28
 */
public class ElasticJobInit implements EnvironmentAware, ApplicationContextAware {

    private static final Logger logger = LoggerFactory.getLogger(ElasticJobInit.class);

    private final static String PLACEHOLDER_PREFIX = "${";

    private final static String PLACEHOLDER_SUFFIX = "}";

    private final ZookeeperRegistryCenter regCenter;

    private ConfigurableEnvironment environment;

    private ApplicationContext applicationContext;


    public ElasticJobInit(ZookeeperRegistryCenter regCenter) {
        this.regCenter = regCenter;
    }

    /**
     * 初始化定时任务
     */
    public void init() {
        String[] jobBeanNameArr = applicationContext.getBeanNamesForAnnotation(ElasticJobConfig.class);
        logger.info("Initializing jobs {}", Arrays.toString(jobBeanNameArr));
        if (jobBeanNameArr.length <= 0) {
            return;
        }
        for (String jobBeanName : jobBeanNameArr) {
            Object jobBean = applicationContext.getBean(jobBeanName);
            Annotation[] beanAnnotation = jobBean.getClass().getAnnotations();
            for (Annotation annotation : beanAnnotation) {
                if (annotation instanceof ElasticJobConfig) {
                    ElasticJobConfig jobConfig = (ElasticJobConfig) annotation;
                    final LiteJobConfiguration.Builder builder = jobConfigurationBuilder(jobConfig, jobBeanName);
                    boolean overwrite = Boolean.parseBoolean(parseStringValue(jobConfig.overwrite()));
                    final LiteJobConfiguration configuration = builder.overwrite(overwrite).build();
                    final SpringJobScheduler scheduler = new SpringJobScheduler((ElasticJob) jobBean, regCenter, configuration);
                    scheduler.init();
                }
            }
        }
    }

    private LiteJobConfiguration.Builder jobConfigurationBuilder(ElasticJobConfig jobConfig, String jobBeanName) {
        //required argument
        String cron = parseStringValue(jobConfig.cron());
        int shardingTotalCount = Integer.parseInt(parseStringValue(jobConfig.shardingTotalCount()));

        //optional argument
        String jobParameter = parseStringValue(jobConfig.jobParameter());
        String description = parseStringValue(jobConfig.description());
        String shardingItemParameters = parseStringValue(jobConfig.shardingItemParameters());

        final JobCoreConfiguration coreConfiguration = JobCoreConfiguration
                .newBuilder(jobBeanName, cron, shardingTotalCount)
                .shardingItemParameters(shardingItemParameters)
                .jobParameter(jobParameter)
                .description(description)
                .build();

        switch (jobConfig.jobType()) {
            case "simple":
                return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(coreConfiguration, jobBeanName));
            case "dataflow":
                return LiteJobConfiguration.newBuilder(new DataflowJobConfiguration(coreConfiguration, jobBeanName, false));
            case "script":
                return LiteJobConfiguration.newBuilder(new ScriptJobConfiguration(coreConfiguration, jobBeanName));
            default:
                throw new IllegalArgumentException("elastic job only support simple|dataflow|script type job");
        }
    }

    public String parseStringValue(String key) {
        if (key != null && !"".equals(key)) {
            int startIndex = key.indexOf(PLACEHOLDER_PREFIX);
            if (startIndex != -1) {
                int endIndex = key.indexOf(PLACEHOLDER_SUFFIX, startIndex + PLACEHOLDER_PREFIX.length());
                if (endIndex != -1) {
                    String parseKey = key.substring(startIndex + PLACEHOLDER_PREFIX.length(), endIndex);
                    return environment.getProperty(parseKey);
                }
            }
        }
        return key;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = (ConfigurableEnvironment) environment;
    }
}
